package org.example;

import java.io.IOException;
import java.util.HashMap;
import java.util.List;
import java.util.Map;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.*;
import org.apache.hadoop.hbase.client.*;
import org.apache.hadoop.hbase.util.Bytes;

public class HBaseJavaClient {

    /**
     * hbase操作必备
     *
     * @return
     */
    private static Configuration getConfiguration() {
        // 获得HBase配置
        Configuration conf = HBaseConfiguration.create();
        // 连接HBase
        conf.set("hbase.rootdir", "hdfs://192.168.134.130:9000/hbase");
        // 使用eclipse时必须添加这个，否则无法定位
        conf.set("hbase.zookeeper.quorum", "192.168.134.130");
        return conf;
    }

    /**
     * 创建一张表
     *
     * @param tableName
     * @param columnFamily
     * @throws IOException
     */
    public static void create(String tableName, String columnFamily) throws IOException {
        // 获取HBase的管理权
        HBaseAdmin admin = new HBaseAdmin(getConfiguration());
        if (admin.tableExists(tableName)) {
            System.out.println("table exists!");
        } else {
            // 创建Table描述器
            HTableDescriptor tableDesc = new HTableDescriptor(tableName);
            // 创建列描述器
            HColumnDescriptor hcd = new HColumnDescriptor(columnFamily);
            // 添加列族
            tableDesc.addFamily(hcd);
            // 建表
            admin.createTable(tableDesc);
            System.out.println("新建表" + tableName + "成功");
        }
    }

    /**
     * 删除表
     *
     * @param tableName
     */
    public static void dropTable(String tableName) {
        //先disable表，然后才可以删除
        try {
            HBaseAdmin admin = new HBaseAdmin(getConfiguration());
            admin.disableTable(tableName);
            admin.deleteTable(tableName);
            System.out.println("表" + tableName + "删除成功");
        } catch (IOException e) {
            throw new RuntimeException(e);
        }

    }

    /**
     * 添加数据
     *
     * @param tableName    表明
     * @param rowKey       行键
     * @param columnFamily 列族
     * @param column       列
     * @param data         数据
     */
    public static void putData(String tableName, String rowKey,
                               String columnFamily, String column, String data) {
        try {
            // 获取表
            HTable table = new HTable(getConfiguration(), tableName);
            //将数据封装成一个put对象
            Put p = new Put(Bytes.toBytes(rowKey));
            //添加列族 列名 值
            p.add(Bytes.toBytes(columnFamily), Bytes.toBytes(column), Bytes.toBytes(data));
            //添加数据到表中
            table.put(p);
            System.out.println("添加数据成功");
        } catch (IOException e) {
            throw new RuntimeException(e);
        }
    }

    /**
     * 查询数据
     *
     * @param tableName 表名
     * @param rowKey    行键
     * @param family    列族
     * @param qualifier
     */
    public static void getData(String tableName, String rowKey,
                               String family, String qualifier) {
        try {
            HTable table = new HTable(getConfiguration(), tableName);
            //通过行间获取
            Get g = new Get(Bytes.toBytes(rowKey));
            //查询结果
            Result result = table.get(g);
            //获取对应的值
            byte[] resultValue = result.getValue(Bytes.toBytes(family), Bytes.toBytes(qualifier));
            System.out.println("获取到：" + Bytes.toString(resultValue));
        } catch (IOException e) {
            throw new RuntimeException(e);
        }
    }

    public static void scan(String tableName) {
        try {
            HTable table = new HTable(getConfiguration(), tableName);
            //创建扫描器
            Scan scan = new Scan();
            //进行扫描，返回的结果可能不止一条
            ResultScanner results = table.getScanner(scan);
            //遍历扫描的结果
            for (Result result : results) {
                byte[] value = result.getValue("apics".getBytes(), "newc".getBytes());
                if (value != null)
                    System.out.println(Bytes.toString(value));
                System.out.println("查询到的结果：" + result);
            }
        } catch (IOException e) {
            throw new RuntimeException(e);
        }

    }

    /**
     * 根据表名、行键、列族获取一个列族的值
     *
     * @param tableName
     * @param rowKey
     * @param family
     * @return
     */
    public static Map<String, String> byGet(String tableName, String rowKey, String family) {
        Map<String, String> result = null;
        try {
            //获取表
            HTable table = new HTable(getConfiguration(), tableName);
            //创建get对象
            Get g = new Get(Bytes.toBytes(rowKey));
            g.addFamily(Bytes.toBytes(family));
            //获取数据
            Result r = table.get(g);
            List<Cell> cells = r.listCells();
            result = cells.size() > 0 ? new HashMap<String, String>() : result;
            for (Cell cell : cells) {
                result.put(
                        Bytes.toString(CellUtil.cloneQualifier(cell)),
                        Bytes.toString(CellUtil.cloneValue(cell))
                );
                System.out.println(result);
            }
        } catch (IOException e) {
            throw new RuntimeException(e);
        }
        return result;
    }


    /**
     * @param args
     * @throws Exception
     */
    public static void main(String[] args) throws Exception {
//        create("apitable", "apics");
//        putData("apitable", "NewUser", "apics", "newc", "Jerry");
//        getData("apitable", "NewUser", "apics", "newc");
//        scan("apitable");
//        dropTable("apitable");
        byGet("apitable", "NewUser", "apics");
    }
}

